package com.shujia.flink.tf;

import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class Demo05Reduce {
    public static void main(String[] args) throws Exception {

        // 用于对KeyBy之后的数据流进行聚合计算
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStream<String> ds = env.socketTextStream("master", 8888);

        // 统计班级的平均年龄
        /*
         * 文科一班,20
         * 文科一班,22
         * 文科一班,21
         * 文科一班,20
         * 文科一班,22
         *
         * 理科一班,20
         * 理科一班,21
         * 理科一班,20
         * 理科一班,21
         * 理科一班,20
         *
         */
        SingleOutputStreamOperator<Tuple3<String, Integer, Integer>> kvDS = ds.map(line -> {
            String[] split = line.split(",");
            String clazz = split[0];
            int age = Integer.parseInt(split[1]);
            return Tuple3.of(clazz, age, 1);
        }, Types.TUPLE(Types.STRING, Types.INT, Types.INT));

        KeyedStream<Tuple3<String, Integer, Integer>, String> keyByDS = kvDS.keyBy(kv -> kv.f0, Types.STRING);

        keyByDS.reduce(new ReduceFunction<Tuple3<String, Integer, Integer>>() {
                    @Override
                    public Tuple3<String, Integer, Integer> reduce(Tuple3<String, Integer, Integer> value1, Tuple3<String, Integer, Integer> value2) throws Exception {
                        return Tuple3.of(value1.f0, value1.f1 + value2.f1, value1.f2 + value2.f2);
                    }
                }).map(t3 -> Tuple2.of(t3.f0, (double) t3.f1 / t3.f2),Types.TUPLE(Types.STRING,Types.DOUBLE))
                .print();

        keyByDS.reduce((v1,v2)->Tuple3.of(v1.f0, v1.f1 + v2.f1, v1.f2 + v2.f2)).print();

        env.execute();
    }
}
